#pragma once
#include <map>
#include <thread>
#include <future>
#include <memory>
#include <queue>
#include <functional>
#include <atomic>
#include <queue>
#include <mutex>
#include <condition_variable>

namespace dvo{

struct signal_lock{
	signal_lock() : condition_(), mutex_(), waiting_(false), busy_(false) { }
    signal_lock(signal_lock&& signal) {}
    void Wait() { std::unique_lock<std::mutex> lk(mutex_); waiting_.store(true); condition_.wait(lk, [&]{ return waiting_.load() == false; }); }
	bool IsWaiting() { return waiting_.load(); }
    bool IsBusy() { return busy_.load(); }
    void SetBusy(bool value) {busy_.store(value); }
	void NotifyAll() { waiting_.store(false); condition_.notify_all(); }
    void NotifyOne() { waiting_.store(false); condition_.notify_one(); }

private:
	std::condition_variable condition_;
	std::mutex mutex_;
	std::atomic<bool> waiting_;
    std::atomic<bool> busy_;
};

class barrier{
private:
    std::mutex mutex_;
    std::condition_variable condition_;
    std::size_t count_;
    std::function<void()> process_;
public:
    explicit barrier(std::size_t count, std::function<void()> process) : count_(count), process_(process) { }
    inline void increase(){ std::unique_lock<std::mutex> lock(mutex_); ++count_;}
    inline void decrease(){ std::unique_lock<std::mutex> lock(mutex_); --count_;}
    void Wait()
    {
        std::unique_lock<std::mutex> lock(mutex_);
        if(--count_ > 0){
          condition_.wait(lock, [this] { return count_ == 0; });
          return;
        }
        condition_.notify_all();
        process_();            
    }
};

namespace ps
{
	template<typename T> class ISimplex{
	public:
		virtual void Transmit(T&) = 0;
        virtual void Transmit(T&&) = 0;
	};
    template<typename T>
    T& operator>>(T& value, ISimplex<T>& target)
    {
      target.Transmit(value);
      return value;
    }
    template<typename T>
    void operator>>(T&& value, ISimplex<T>& target)
    {
      target.Transmit(value);
    }
	template<typename T> class Socket : public ISimplex < T > {
	public:
      Socket<T>& operator=(std::function<void(T&)> func) { Bind(func); return *this; }
		void Bind(std::function<void(T&)> func){ transmitter_ = func; }
		virtual void Transmit(T& value) override { transmitter_(value); }
        virtual void Transmit(T&& value) override { transmitter_(value); }
	private:
		std::function<void(T&)> transmitter_;
	};
    template<typename T>
    void operator>>(Socket<T>& socket, std::function<void(T&)> func){
      socket.Bind(func);
    }
	template<typename TKey, typename TValue> class KeyedSocket : public ISimplex < TValue > {
	public:
		KeyedSocket(TKey key) : key_(key) { }
		void Bind(std::function<void(TKey, TValue&)> func){ transmitter_ = func; }
		virtual void Transmit(TValue& value) override { transmitter_(key_, value); }
        virtual void Transmit(TValue&& value) override {  transmitter_(key_, value); }
	private:
		TKey key_;
		std::function<void(TKey, TValue&)> transmitter_;
	};
    template<typename TKey, typename TValue>
    void operator>>(KeyedSocket<TKey, TValue>& socket, std::function<void(TValue&)> func){
      socket.Bind(func);
    }
	template<typename T> class Plug : public ISimplex < T > {
	public:
		inline void Connect(ISimplex<T>* target) { target_ = target; }
		virtual void Transmit(T& value) override { target_->Transmit(value); }
        virtual void Transmit(T&& value) override { target_->Transmit(value); }
	private:
		ISimplex<T>* target_;
	};
    template<typename T>
    Plug<T>& operator>>(Plug<T>& plug, Plug<T>* target){
      plug.Connect(target);
      return *target;
    }
    template<typename T>
    void operator>>(Plug<T>& plug, ISimplex<T>* target){
      plug.Connect(target);
    }
	//---------------------------------------------------------------------------------

	template<typename T> class Receiver {
	public:
		Receiver();
		virtual ~Receiver();
		Socket<T>* Input(){ return input_.get(); }
	protected:
		virtual void Process(T& value) = 0;
	private:
		std::unique_ptr<Socket<T>> input_;
	};

	template<typename T>
	Receiver<T>::Receiver()
	{
		input_.reset(new Socket<T>());
		input_->Bind(std::bind(&Receiver<T>::Process, this, std::placeholders::_1));
	}

	template<typename T>
	Receiver<T>::~Receiver()
	{

	}

    //---------------------------------------------------------------------------------
    template<typename T>
    class SimpleReceiver : public Receiver<T>
    {
    public:
      SimpleReceiver& operator=(std::function<void(T&)> process) { process_ = process; return *this; }
      SimpleReceiver(std::function<void(T&)> process) { process_ = process; }
      virtual ~SimpleReceiver(){}
    protected:
      virtual void Process(T& value) override {process_(value);}
    private:
      std::function<void(T&)> process_;
    };

	//---------------------------------------------------------------------------------

	template<typename T> class Transmitter {
	public:
		Transmitter();
		virtual ~Transmitter();
		Plug<T>* Output(){ return output_.get(); }
        virtual void Process() = 0;
	protected:
		inline void Send(T& value) { output_->Transmit(value); }
	private:
		std::unique_ptr<Plug<T>> output_;
	};

	template<typename T>
	Transmitter<T>::Transmitter()
	{
      output_.reset(new Plug<T>());
	}

	template<typename T>
	Transmitter<T>::~Transmitter()
	{

	}

    //---------------------------------------------------------------------------------
    template<typename T>
    class SimpleTransmitter : public Transmitter<T>
    {
    public:
      SimpleTransmitter& operator=(std::function<T(void)> process) { process_ = process; return *this; }
      SimpleTransmitter(std::function<T(void)> process) { process_ = process; }
      virtual ~SimpleTransmitter(){}
      virtual void Process() override { Send(process_());}
    private:
      std::function<T(void)> process_;
    };

	//---------------------------------------------------------------------------------

	template<typename TInput, typename TOutput> class Transceiver {
	public:
		Transceiver();
		virtual ~Transceiver();
		Socket<TInput>* Input(){ return input_.get(); }
		Plug<TOutput>* Output(){ return output_.get(); }
	protected:
		virtual void Process(TInput& value) = 0;
		inline void Send(TOutput& value) { output_->Transmit(value); }
        inline void Send(TOutput&& value) { output_->Transmit(value); }
	private:
		std::unique_ptr<Socket<TInput>> input_;
		std::unique_ptr<Plug<TOutput>> output_;
	};

	template<typename TInput, typename TOutput>
	Transceiver<TInput, TOutput>::Transceiver()
	{
		input_.reset(new Socket<TInput>());
		output_.reset(new Plug<TOutput>());
        input_->Bind(std::bind(&Transceiver<TInput, TOutput>::Process, this, std::placeholders::_1));
	}

	template<typename TInput, typename TOutput>
	Transceiver<TInput, TOutput>::~Transceiver()
	{

	}
    
    //---------------------------------------------------------------------------------
    template<typename TInput, typename TOutput>
    class SimpleTransceiver : public Transceiver<TInput, TOutput>
    {
    public:
      SimpleTransceiver& operator=(std::function<TOutput(TInput&)> process) { process_ = process; return *this; }
      SimpleTransceiver(std::function<TOutput(TInput&)> process) { process_ = process; }
      virtual ~SimpleTransceiver(){}
    protected:
      virtual void Process(TInput& value) override {Send(std::forward<TOutput>(process_(value)));}
    private:
      std::function<TOutput(TInput&)> process_;
    };

	//---------------------------------------------------------------------------------
		
	template<typename TKey, typename TInput, typename TOutput> class Joiner {
	public:
		Joiner();
		virtual ~Joiner();
		KeyedSocket<TKey, TInput>* Inputs(TKey key){ return inputs_[key].get(); }
		Plug<TOutput>* Output(){ return output_.get(); }
	protected:
        virtual void Add(TKey key);
        virtual void Remove(TKey key);
		virtual void Process() = 0;
		inline void Send(TOutput& value) { output_->Transmit(value); }
		virtual void Set(TKey key, TInput& value);
        virtual TInput Get(TKey);
	private:
		std::map<TKey, std::unique_ptr<KeyedSocket<TKey, TInput>>> inputs_;
		std::map<TKey, TInput> values_;
        std::unique_ptr<barrier> barrier_;
		std::unique_ptr<Plug<TOutput>> output_;
	};

	template<typename TKey, typename TInput, typename TOutput>
	Joiner<TKey, TInput, TOutput>::Joiner()
	{
		output_.reset(new Plug<TOutput>());
        barrier_.reset(new barrier(0, [&](){async(std::bind(&Joiner<TKey, TInput, TOutput>::Process, this)); }));
	}

	template<typename TKey, typename TInput, typename TOutput>
	void Joiner<TKey, TInput, TOutput>::Add(TKey key)
	{
		barrier_->increase();            
        inputs_.emplace(key, std::unique_ptr<KeyedSocket<TKey, TInput>>(new KeyedSocket<TKey, TInput>(key)));
		inputs_[key]->Bind(std::bind(&Joiner<TKey, TInput, TOutput>::Set, this, std::placeholders::_1, std::placeholders::_2));
	}

    template<typename TKey, typename TInput, typename TOutput>
	void Joiner<TKey, TInput, TOutput>::Remove(TKey key)
	{
		barrier_->decrease();            
        inputs_.erase(key);
	}

	template<typename TKey, typename TInput, typename TOutput>
	void Joiner<TKey, TInput, TOutput>::Set(TKey key, TInput& value)
	{
		values_[key] = std::move(value);
        barrier_->Wait();
	}

    template<typename TKey, typename TInput, typename TOutput>
	TInput Joiner<TKey, TInput, TOutput>::Get(TKey key)
	{
		return values_[key];
	}

	template<typename TKey, typename TInput, typename TOutput>
	Joiner<TKey, TInput, TOutput>::~Joiner()
	{

	}

   
	//---------------------------------------------------------------------------------

	template<typename TKey, typename TInput, typename TOutput> class Detacher {
	public:
		Detacher();
		virtual ~Detacher();
		Socket<TInput>* Input(){ return input_.get(); }
		Plug<TOutput>* Outputs(TKey key){ return outputs_[key].get(); }
	protected:
		virtual void Add(TKey key);
        virtual void Remove(TKey key);
		virtual std::future<void> Send(TKey key, TOutput& value);
		virtual void Process(TInput& value) = 0;
	private:
		std::unique_ptr<Socket<TInput>> input_;
		std::map<TKey, std::unique_ptr<Plug<TOutput>>> outputs_;
        std::map<TKey, TOutput> values_;
	};

	template<typename TKey, typename TInput, typename TOutput>
	Detacher<TKey, TInput, TOutput>::Detacher()
	{
		input_.reset(new Socket<TInput>());
		input_->Bind(std::bind(&Detacher<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
	}

	template<typename TKey, typename TInput, typename TOutput>
	void Detacher<TKey, TInput, TOutput>::Add(TKey key)
	{
		outputs_.emplace(key, std::unique_ptr<Plug<TOutput>>(new Plug<TOutput>()));
	}

    template<typename TKey, typename TInput, typename TOutput>
	void Detacher<TKey, TInput, TOutput>::Remove(TKey key)
	{
		outputs_.erase(key);
	}

	template<typename TKey, typename TInput, typename TOutput>
	std::future<void> Detacher<TKey, TInput, TOutput>::Send(TKey key, TOutput& value)
	{
		Plug<TOutput> *output = outputs_[key].get();
        values_[key] = std::move(value);
		return std::async(std::launch::async, [&](TKey forwarded) { outputs_[forwarded]->Transmit(values_[forwarded]); }, key);
	}

	template<typename TKey, typename TInput, typename TOutput>
	Detacher<TKey, TInput, TOutput>::~Detacher()
	{

	}

	//---------------------------------------------------------------------------------

	template<typename TKey, typename TInput, typename TOutput> class Selector {
	public:
		Selector();
		virtual ~Selector();
		Socket<TInput>* Input(){ return input_.get(); }
		Plug<TOutput>* Outputs(TKey key){ return outputs_[key].get(); }
	protected:
		virtual void Add(TKey key);
		inline void Send(TKey key, TOutput& value) { outputs_[key]->Transmit(value); }
		virtual void Process(TInput& value) = 0;
	private:
		std::unique_ptr<Socket<TInput>> input_;
		std::map<TKey, std::unique_ptr<Plug<TOutput>>> outputs_;
	};

	template<typename TKey, typename TInput, typename TOutput>
	Selector<TKey, TInput, TOutput>::Selector()
	{
		input_.reset(new Socket<TInput>());
		input_->Bind(std::bind(&Selector<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
	}

	template<typename TKey, typename TInput, typename TOutput>
	void Selector<TKey, TInput, TOutput>::Add(TKey key)
	{
		outputs_.emplace(key, std::unique_ptr<Plug<TOutput>>(new Plug<TOutput>()));
	}

	template<typename TKey, typename TInput, typename TOutput>
	Selector<TKey, TInput, TOutput>::~Selector()
	{

	}

	//---------------------------------------------------------------------------------

	template<typename TKey, typename TInput, typename TOutput> class Restrictor {
	public:
		Restrictor();
		virtual ~Restrictor();
		KeyedSocket<TKey, TInput>* Inputs(TKey key){ return inputs_[key].get(); }
		Plug<TOutput>* Output(){ return output_.get(); }
	protected:
		virtual void Add(TKey key);
		virtual void Process(TKey, TInput&) = 0;
		inline void Send(TOutput& value) { output_->Transmit(value); }
	private:
		std::map<TKey, std::unique_ptr<KeyedSocket<TKey, TInput>>> inputs_;
		std::unique_ptr<Plug<TOutput>> output_;
	};

	template<typename TKey, typename TInput, typename TOutput>
	Restrictor<TKey, TInput, TOutput>::Restrictor()
	{
		output_.reset(new Plug<TOutput>());
	}

	template<typename TKey, typename TInput, typename TOutput>
	void Restrictor<TKey, TInput, TOutput>::Add(TKey key)
	{
		inputs_.emplace(key, std::unique_ptr<KeyedSocket<TKey, TInput>>(new KeyedSocket<TKey, TInput>(key)));
		inputs_[key]->Bind(std::bind(&Restrictor<TKey, TInput, TOutput>::Process, this, std::placeholders::_1, std::placeholders::_2));
	}

	template<typename TKey, typename TInput, typename TOutput>
	Restrictor<TKey, TInput, TOutput>::~Restrictor()
	{

	}

//---------------------------------------------------------------------------------

template<typename T> class Separator {
public:
	Separator();
	virtual ~Separator();
	Socket<T>* Input(){ return input_.get(); }
	Plug<T>* Output(){ return output_.get(); }
protected:
	virtual void Receive(T& value);
    virtual void Process();
    inline void Send(T& value) {output_->Transmit(value);}
private:
	std::unique_ptr<Socket<T>> input_;
	std::unique_ptr<Plug<T>> output_;
	T value_;
    signal_lock signal_;
};

template<typename T>
Separator<T>::Separator()
{
	input_.reset(new Socket<T>());
	input_->Bind(std::bind(&Separator<T>::Receive, this, std::placeholders::_1));
    output_.reset(new Plug<T>());
}

template<typename T>
void Separator<T>::Receive(T& value)
{
  if(signal_.IsBusy()){
    signal_.Wait();
  }
  signal_.SetBusy(true);
  value_ = std::move(value);
  std::async(std::launch::async, std::bind(&Separator<T>::Process,this));
}

template<typename T>
void Separator<T>::Process()
{
  Send(value_);
  if(signal_.IsWaiting()){
    signal_.NotifyOne();
  }
  signal_.SetBusy(false);
}

template<typename T>
Separator<T>::~Separator()
{
  
}
}//end of Pushflow namespace


namespace pl
{
template<typename T> class ISimplex{
public:
	virtual T Request() = 0;
};
template<typename T>
T& operator<<(T& target, ISimplex<T>& source){
  target = source.Request();
  return target;
}
template<typename T> class Socket : public ISimplex < T > {
public:
    Socket<T>& operator=(std::function<T(void)> func) { Bind(func); return *this; }
	void Bind(std::function<T(void)> func){ requester_ = func; }
	virtual T Request() override { return requester_(); }
private:
	std::function<T(void)> requester_;
};
template<typename T>
void operator<<(Socket<T>& socket, std::function<T(void)> func){
  socket.Bind(func);
}
template<typename TKey, typename TValue> class KeyedSocket : public ISimplex < TValue > {
public:
	KeyedSocket(TKey key) : key_(key) { }
	void Bind(std::function<TValue(TKey)> func){ requester_ = func; }
	virtual TValue Request() override { return requester_(key_); }
private:
	TKey key_;
	std::function<TValue(TKey)> requester_;
};
template<typename TKey, typename TValue>
void operator<<(KeyedSocket<TKey, TValue>& socket, std::function<TValue(void)> func){
  socket.Bind(func);
}
template<typename T> class Plug : public ISimplex<T> {
public:
	inline void Connect(ISimplex<T>* target) { target_ = target; }
	virtual T Request() override { return target_->Request(); }
private:
	ISimplex<T>* target_;
};
template<typename T>
Plug<T>& operator<<(Plug<T>& target, Plug<T>* source){
  source->Connect(&target);
  return target;
}
template<typename T>
void operator<<(ISimplex<T>& target, Plug<T>* source){
  source->Connect(&target);
}
//---------------------------------------------------------------------------------

template<typename T> class Receiver {
public:
  Receiver();
  virtual ~Receiver();
  Plug<T>* Input(){ return input_.get(); }
protected:
  virtual void Process() = 0;
  inline T Demand() { return input_->Request(); }
private:
  std::unique_ptr<Plug<T>> input_;
};

template<typename T>
Receiver<T>::Receiver()
{
  input_.reset(new Plug<T>());
}

template<typename T>
Receiver<T>::~Receiver()
{

}

//---------------------------------------------------------------------------------
template<typename T> class Transmitter {
public:
  Transmitter();
  virtual ~Transmitter();
  Socket<T>* Output(){ return output_.get(); }
protected:
  virtual T Process() = 0;
private:
	std::unique_ptr<Socket<T>> output_;
};

template<typename T>
Transmitter<T>::Transmitter()
{
  output_.reset(new Socket<T>());
  output_->Bind(std::bind(&Transmitter<T>::Process, this));
}

template<typename T>
Transmitter<T>::~Transmitter()
{
  output_.reset();
}
//---------------------------------------------------------------------------------
template<typename T>
class SimpleTransmitter : public Transmitter<T>
{
public:
  SimpleTransmitter& operator=(std::function<T()> process) { process_ = process; return *this; }
  SimpleTransmitter(std::function<T()> process) { process_ = process; }
  virtual ~SimpleTransmitter(){}
protected:
  virtual T Process() override {return process_();}
private:
  std::function<T()> process_;
};
//---------------------------------------------------------------------------------
template<typename TInput, typename TOutput> class Transceiver {
public:
  Transceiver();
  virtual ~Transceiver();
  Plug<TInput>* Input(){ return input_.get(); }
  Socket<TOutput>* Output(){ return output_.get(); }
protected:
  inline TInput Demand() { return input_.get()->Request(); }
  virtual TOutput Process() = 0;
private:
  std::unique_ptr<Plug<TInput>> input_;
  std::unique_ptr<Socket<TOutput>> output_;
};

template<typename TInput, typename TOutput>
Transceiver<TInput, TOutput>::Transceiver()
{
  input_.reset(new Plug<TInput>());
  output_.reset(new Socket<TOutput>());
  output_->Bind(std::bind(&Transceiver<TInput, TOutput>::Process, this));
}

template<typename TInput, typename TOutput>
Transceiver<TInput, TOutput>::~Transceiver()
{

};
//---------------------------------------------------------------------------------
template<typename TInput, typename TOutput>
class SimpleTransceiver : public Transceiver<TInput, TOutput>
{
public:
  SimpleTransceiver& operator=(std::function<TOutput(TInput&)> process) { process_ = process; return *this; }
  SimpleTransceiver(std::function<TOutput(TInput&)> process) { process_ = process; }
  virtual ~SimpleTransceiver(){}
protected:
  virtual TOutput Process() override {return process_(Demand());}
private:
  std::function<TOutput(TInput&)> process_;
};
//---------------------------------------------------------------------------------

template<typename TKey, typename TInput, typename TOutput> class Joiner {
public:
	Joiner();
	virtual ~Joiner();
	Plug<TInput>* Input(){ return input_.get(); }
    KeyedSocket<TKey, TOutput>* Outputs(TKey key){ return outputs_[key].get(); }
protected:
	virtual void Add(TKey key);
    virtual void Remove(TKey key);
	virtual TOutput Supply(TKey key);
    inline TInput Demand() { return input_->Request(); }
	virtual TOutput Process() = 0;
private:
	std::map<TKey, std::shared_ptr<KeyedSocket<TKey, TOutput>>> outputs_;
    std::unique_ptr<Plug<TInput>> input_;
    std::unique_ptr<barrier> barrier_;
};

template<typename TKey, typename TInput, typename TOutput>
Joiner<TKey, TInput, TOutput>::Joiner()
{
    input_.reset(new Plug<TInput>());
    barrier_.reset(new barrier(0, [](){}));
}

template<typename TKey, typename TInput, typename TOutput>
void Joiner<TKey, TInput, TOutput>::Add(TKey key)
{
	outputs_.emplace(key, std::make_shared<KeyedSocket<TKey, TOutput>>(key));
	outputs_[key]->Bind(std::bind(&Joiner<TKey, TInput, TOutput>::Supply, this, std::placeholders::_1));
    barrier_->increase();
}

template<typename TKey, typename TInput, typename TOutput>
void Joiner<TKey, TInput, TOutput>::Remove(TKey key)
{
  outputs_.erase(key);
  barrier_->decrease();
}


template<typename TKey, typename TInput, typename TOutput>
TOutput Joiner<TKey, TInput, TOutput>::Supply(TKey key)
{
	TOutput result = Process();
    barrier_->Wait();
	return result;
}

template<typename TKey, typename TInput, typename TOutput>
Joiner<TKey, TInput, TOutput>::~Joiner()
{
    input_.reset();
}


//---------------------------------------------------------------------------------

template<typename TKey, typename TInput, typename TOutput> class Detacher {
public:
	Detacher();
	virtual ~Detacher();
    Plug<TInput>* Inputs(TKey key){ return inputs_[key].get(); }
    Socket<TOutput>* Output(){ return output_.get(); }
protected:
	virtual void Add(TKey key);
    virtual void Remove(TKey key);
	virtual std::future<TInput> Demand(TKey key);
	virtual TOutput Process() = 0;
private:
	std::map<TKey, std::unique_ptr<Plug<TInput>>> inputs_;
    std::unique_ptr<Socket<TOutput>> output_;
};

template<typename TKey, typename TInput, typename TOutput>
Detacher<TKey, TInput, TOutput>::Detacher()
{
    output_.reset(new Socket<TInput>());
    output_->Bind(std::bind(&Detacher<TKey, TInput, TOutput>::Process, this));
}

template<typename TKey, typename TInput, typename TOutput>
void Detacher<TKey, TInput, TOutput>::Add(TKey key)
{
	inputs_.emplace(key, std::unique_ptr<Plug<TInput>>(new Plug<TInput>()));
}

template<typename TKey, typename TInput, typename TOutput>
void Detacher<TKey, TInput, TOutput>::Remove(TKey key)
{
	inputs_.erase(key);
}

template<typename TKey, typename TInput, typename TOutput>
std::future<TInput> Detacher<TKey, TInput, TOutput>::Demand(TKey key)
{
  return std::async(std::launch::async, [&](TKey forwarded) { return inputs_[forwarded]->Request(); }, key);
}

template<typename TKey, typename TInput, typename TOutput>
Detacher<TKey, TInput, TOutput>::~Detacher()
{
	output_.reset();
}



//---------------------------------------------------------------------------------

template<typename TKey, typename TInput, typename TOutput> class Selector {
public:
	Selector();
	virtual ~Selector();
    Plug<TInput>* Inputs(TKey key){ return inputs_[key].get(); }
    Socket<TOutput>* Output(){ return output_.get(); }
protected:
	virtual void Add(TKey key);
    virtual void Remove(TKey key);
	virtual TInput Demand(TKey key);
	virtual TOutput Process() = 0;
private:
	std::map<TKey, std::unique_ptr<Plug<TInput>>> inputs_;
    std::unique_ptr<Socket<TOutput>> output_;
};

template<typename TKey, typename TInput, typename TOutput>
Selector<TKey, TInput, TOutput>::Selector()
{
    output_.reset(new Socket<TInput>());
    output_->Bind(std::bind(&Selector<TKey, TInput, TOutput>::Process, this));
}

template<typename TKey, typename TInput, typename TOutput>
void Selector<TKey, TInput, TOutput>::Add(TKey key)
{
	inputs_.emplace(key, std::unique_ptr<Plug<TInput>>(new Plug<TInput>()));
}

template<typename TKey, typename TInput, typename TOutput>
void Selector<TKey, TInput, TOutput>::Remove(TKey key)
{
	inputs_.erase(key);
}

template<typename TKey, typename TInput, typename TOutput>
TInput Selector<TKey, TInput, TOutput>::Demand(TKey key)
{
  return inputs_[key]->Request();
}

template<typename TKey, typename TInput, typename TOutput>
Selector<TKey, TInput, TOutput>::~Selector()
{
	output_.reset();
}


//---------------------------------------------------------------------------------

template<typename TKey, typename TInput, typename TOutput> class Restrictor {
public:
	Restrictor();
	virtual ~Restrictor();
	Plug<TInput>* Input(){ return input_.get(); }
    KeyedSocket<TKey, TOutput>* Outputs(TKey key){ return outputs_[key].get(); }
protected:
	virtual void Add(TKey key);
    virtual void Remove(TKey key);
    inline TInput Demand() { return input_.get()->Request(); }
	virtual TOutput Process(TKey key) = 0;
private:
	std::map<TKey, std::unique_ptr<KeyedSocket<TKey, TOutput>>> outputs_;
    std::unique_ptr<Plug<TInput>> input_;
};

template<typename TKey, typename TInput, typename TOutput>
Restrictor<TKey, TInput, TOutput>::Restrictor()
{
    input_.reset(new Plug<TInput>());
}

template<typename TKey, typename TInput, typename TOutput>
void Restrictor<TKey, TInput, TOutput>::Add(TKey key)
{
	outputs_.emplace(key, std::unique_ptr<KeyedSocket<TKey, TOutput>>(new KeyedSocket<TKey, TOutput>(key)));
	outputs_[key]->Bind(std::bind(&Restrictor<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
}

template<typename TKey, typename TInput, typename TOutput>
void Restrictor<TKey, TInput, TOutput>::Remove(TKey key)
{
  outputs_.erase(key);
}

template<typename TKey, typename TInput, typename TOutput>
Restrictor<TKey, TInput, TOutput>::~Restrictor()
{

}

//---------------------------------------------------------------------------------

template<typename T> class Separator {
public:
	Separator();
	virtual ~Separator();
	Plug<T>* Input(){ return input_.get(); }
	Socket<T>* Output(){ return output_.get(); }
protected:
	virtual T Supply();
    inline T Demand() { return input_->Request(); }
private:
	std::unique_ptr<Plug<T>> input_;
	std::unique_ptr<Socket<T>> output_;
    std::mutex mutex_;
};

template<typename T>
Separator<T>::Separator()
{
	input_.reset(new Plug<T>());
    output_.reset(new Socket<T>());
	output_->Bind(std::bind(&Separator<T>::Supply, this));
}

template<typename T>
T Separator<T>::Supply()
{
  std::lock_guard<std::mutex> guard(mutex_);
  std::future<T> result = std::async(std::launch::async, std::bind(&Separator<T>::Demand, this));
  return result.get();
  
}

template<typename T>
Separator<T>::~Separator()
{

}

}//end of Pullflow namespace

template<typename T>
class Relayer
{
public: 
  Relayer();
  ~Relayer();
  inline pl::Plug<T>* Input() { return input_.get();  }
  inline ps::Plug<T>* Output() { return output_.get(); }
  void Process();
protected:
  inline T Demand() { return input_.get()->Request(); }
  inline void Send(T& value) { output_->Transmit(value); }
private:
  std::unique_ptr<pl::Plug<T>> input_;
  std::unique_ptr<ps::Plug<T>> output_;
};

template<typename T>
Relayer<T>::Relayer()
{
  input_.reset(new pl::Plug<T>());
  output_.reset(new ps::Plug<T>());
}

template<typename T>
Relayer<T>::~Relayer()
{

}

template<typename T>
void Relayer<T>::Process()
{
  T value = Demand();
  Send(value);
}
  
template<typename T>
class Accumulator
{
public: 
  Accumulator(int capacity = 0);
  ~Accumulator();
  inline ps::Socket<T>* Input() { return input_.get();  }
  inline pl::Socket<T>* Output() { return output_.get(); }
protected:
  virtual void Receive(T& value);
  virtual T Supply();
  void Enqueue(T&& value);
  T Dequeue();
  inline const int Size() {return queue_->size();}
private:
  std::unique_ptr<ps::Socket<T>> input_;
  std::unique_ptr<pl::Socket<T>> output_;
  std::unique_ptr<std::queue<T>> queue_;
  const int capacity_;
  signal_lock signal_;
};

template<typename T>
Accumulator<T>::Accumulator(int capacity) : capacity_(capacity)
{
  input_.reset(new ps::Socket<T>());
  output_.reset(new pl::Socket<T>());
  queue_.reset(new std::queue<T>());
  input_->Bind(std::bind(&Accumulator<T>::Receive, this, std::placeholders::_1));
  output_->Bind(std::bind(&Accumulator<T>::Supply, this));
}

    
template<typename T>
Accumulator<T>::~Accumulator()
{

}

template<typename T>
void Accumulator<T>::Receive(T& value)
{
  const int count = queue_->size();
  if(capacity_ > 0 && count >= capacity_){
    signal_.Wait();    
  }
  Enqueue(std::forward<T>(value));
  if(count == 0){
    signal_.NotifyAll();
  }
}

template<typename T>
T Accumulator<T>::Supply()
{
  const int count = queue_->size();
  if(count == 0){
    signal_.Wait();
  }
  T result = Dequeue();
  if(capacity_ > 0 && count >= capacity_){
    signal_.NotifyOne();
  }
  return result;
}

template<typename T>
void Accumulator<T>::Enqueue(T&& value)
{
  queue_->emplace<T>(std::forward<T>(value));
}
template<typename T>
T Accumulator<T>::Dequeue()
{
  T result = std::move(queue_->front());
  queue_->pop();
  return result;
}

}//end of dvo alias devlego namespace